package com.fbs.job.core.thread;

import com.fbs.job.core.biz.model.HandleCallbackParam;
import com.fbs.job.core.biz.model.TriggerParam;
import com.fbs.job.core.context.FbsJobContext;
import com.fbs.job.core.context.FbsJobHelper;
import com.fbs.job.core.executor.FbsJobExecutor;
import com.fbs.job.core.handler.IJobHandler;
import com.fbs.job.core.model.ResultT;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * handler thread
 */
public class JobThread extends Thread{
	private static Logger logger = LoggerFactory.getLogger(JobThread.class);
	private int jobId;
	private IJobHandler handler;

	//触发队列
	private LinkedBlockingQueue<TriggerParam> triggerQueue;
	//触发任务Log ID
	private Set<Long> triggerLogIdSet;		// avoid repeat trigger for the same TRIGGER_LOG_ID

	private volatile boolean toStop = false;

	private String stopReason;

	private boolean running = false;    // if running job

	private int idleTimes = 0;			// idel times
	public JobThread(int jobId, IJobHandler handler) {
		this.jobId = jobId;
		this.handler = handler;
		this.triggerQueue = new LinkedBlockingQueue<TriggerParam>();
		this.triggerLogIdSet = Collections.synchronizedSet(new HashSet<Long>());

		// assign job thread name
		this.setName("fbs-job, JobThread-"+jobId+"-"+System.currentTimeMillis());
	}
	public IJobHandler getHandler()
	{
		return handler;
	}

	public ResultT<String> pushTriggerQueue(TriggerParam triggerParam){
		// avoid repeat
		if (triggerLogIdSet.contains(triggerParam.getLogId())) {
			logger.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId());
			return new ResultT<String>(ResultT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId());
		}
		triggerLogIdSet.add(triggerParam.getLogId());
		triggerQueue.add(triggerParam);

		return ResultT.success();
	}


	/**
	 * kill job thread
	 *
	 * @param stopReason
	 */
	public void toStop(String stopReason) {
		/**
		 * Thread.interrupt只支持终止线程的阻塞状态(wait、join、sleep)，
		 * 在阻塞出抛出InterruptedException异常,但是并不会终止运行的线程本身；
		 * 所以需要注意，此处彻底销毁本线程，需要通过共享变量方式；
		 */
		this.toStop = true;
		this.stopReason = stopReason;
	}

	/**
	 * is running job
	 * @return
	 */
	public boolean isRunningOrHasQueue() {
		return running || triggerQueue.size()>0;
	}

	@Override
	public void run() {
		// init
		try {
			handler.init();
		} catch (Throwable e) {
			logger.error(e.getMessage(), e);
		}
		while (!toStop){
			running = false;
			idleTimes++;
			TriggerParam triggerParam = null;
			try {
				triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
				if (triggerParam!=null){
					running = true;
					idleTimes = 0;
					triggerLogIdSet.remove(triggerParam.getLogId());
					FbsJobContext fbsJobContext = new FbsJobContext(
							triggerParam.getJobId(),
							triggerParam.getExecutorParams(),
							null,
							triggerParam.getBroadcastIndex(),
							triggerParam.getBroadcastTotal());
				// init job context
					FbsJobContext.setFbsJobContext(fbsJobContext);

					if (triggerParam.getExecutorTimeout() > 0){
						// limit timeout
						Thread futureThread = null;
						try {


							FutureTask<Boolean> futureTask=	new FutureTask<Boolean>(new Callable<Boolean>() {
								@Override
								public Boolean call() throws Exception {
									// init job context
									FbsJobContext.setFbsJobContext(fbsJobContext);

									handler.execute();
									return true;
								}
							});
							futureThread = new Thread(futureTask);
							futureThread.start();

							Boolean tempResult = futureTask.get(triggerParam.getExecutorTimeout(), TimeUnit.SECONDS);
						}catch (Exception e){
							// handle result
							FbsJobHelper.handleTimeout("job execute timeout ");
						}finally {
							futureThread.interrupt();
						}
					}else {
						// just execute
						handler.execute();
					}


					// valid execute handle data
					if (FbsJobContext.getFbsJobContext().getHandleCode() <= 0) {
						FbsJobHelper.handleFail("job handle result lost.");
					} else {
						String tempHandleMsg = FbsJobContext.getFbsJobContext().getHandleMsg();
						tempHandleMsg = (tempHandleMsg!=null&&tempHandleMsg.length()>50000)
								?tempHandleMsg.substring(0, 50000).concat("...")
								:tempHandleMsg;
						FbsJobContext.getFbsJobContext().setHandleMsg(tempHandleMsg);
					}

				}else{
					if (idleTimes > 30) {
						if(triggerQueue.size() == 0) {	// avoid concurrent trigger causes jobId-lost
							FbsJobExecutor.removeJobThread(jobId, "excutor idel times over limit.");
						}
					}
				}

			} catch (Exception e){
				if (toStop) {

				}
				// handle result
				StringWriter stringWriter = new StringWriter();
				e.printStackTrace(new PrintWriter(stringWriter));
				String errorMsg = stringWriter.toString();

				FbsJobHelper.handleFail(errorMsg);
			}finally {
				if(triggerParam != null) {
					// callback handler info
					if (!toStop) {
						// commonm
						TriggerCallbackThread.pushCallBack(new HandleCallbackParam(
								triggerParam.getLogId(),
								triggerParam.getLogDateTime(),
								FbsJobContext.getFbsJobContext().getHandleCode(),
								FbsJobContext.getFbsJobContext().getHandleMsg() )
						);
					} else {
						// is killed
						TriggerCallbackThread.pushCallBack(new HandleCallbackParam(
								triggerParam.getLogId(),
								triggerParam.getLogDateTime(),
								FbsJobContext.HANDLE_CODE_FAIL,
								stopReason + " [job running, killed]" )
						);
					}
				}
			}
		}
		// callback trigger request in queue
		while(triggerQueue !=null && triggerQueue.size()>0){
			TriggerParam triggerParam = triggerQueue.poll();
			if (triggerParam!=null) {
				// is killed
				TriggerCallbackThread.pushCallBack(new HandleCallbackParam(
						triggerParam.getLogId(),
						triggerParam.getLogDateTime(),
						FbsJobContext.HANDLE_CODE_FAIL,
						stopReason + " [job not executed, in the job queue, killed.]")
				);
			}
		}

		// destroy
		try {
			handler.destroy();
		} catch (Throwable e) {
			logger.error(e.getMessage(), e);
		}

		logger.info(">>>>>>>>>>> fbs-job JobThread stopped, hashCode:{}", Thread.currentThread());
	}
}
